//! 测试客户端连接超时后服务端的行为是否正常

use quic_rpc_macros::service;
use quic_rpc_utils::{
    QuinnConnector, QuinnEndpoint, QuinnListener, Result, RpcClient, RpcServer, StreamExt,
    configure_client, configure_server, gen_server_cert, run_server,
};
use std::{
    env::{set_var, var},
    net::{Ipv4Addr, SocketAddr, SocketAddrV4},
};
use tokio::{
    spawn,
    time::{Duration, sleep},
};

#[derive(Clone, Default)]
struct TestLostService;

#[service(public = crate, server = true, client = true)]
impl TestLostService {
    async fn test(&self) -> std::result::Result<(), ()> {
        Ok(())
    }

    /// 回显字符串，演示双向流模式
    async fn echo(
        &self,
        name: String,
        flag: u8,
        sender: impl quic_rpc_utils::Stream<Item = u8>,
    ) -> impl quic_rpc_utils::Stream<Item = String> {
        async_stream::stream! {
            quic_rpc_utils::pin!(sender);
            while let Some(item) = sender.next().await {
                yield format!("Echo {}, {}", name, item + flag)
            }

            unsafe { set_var("STREAM_FINISHED", "true") };
        }
    }
}

#[tokio::test]
async fn test_lost() -> Result<()> {
    let server_addr: SocketAddr = SocketAddr::V4(SocketAddrV4::new(Ipv4Addr::LOCALHOST, 9988));
    let (server_cert, private_key) = gen_server_cert(&["localhost"])?;
    let server_config = configure_server(0, server_cert.clone(), private_key)?;
    let server = QuinnEndpoint::server(server_config, server_addr)?;
    let client_cfg = configure_client(&[&server_cert])?;
    let mut client = QuinnEndpoint::client("0.0.0.0:0".parse()?)?;
    client.set_default_client_config(client_cfg);

    let server_conn = QuinnListener::new(server).unwrap();
    let client_conn = QuinnConnector::new(client, server_addr, "localhost".into());

    unsafe { set_var("STREAM_FINISHED", "false") };
    let server = RpcServer::<TestLostService, _>::new(server_conn);
    spawn(run_server(server));
    sleep(Duration::from_millis(1000)).await;
    let client = RpcClient::<TestLostService, _>::new(client_conn);
    sleep(Duration::from_millis(1000)).await;
    let client = TestLostServiceClient::new(&client);
    client.test().await?.unwrap();
    let (mut server_stream, mut client_stream) = client.echo("hello".to_string(), 1).await.unwrap();
    server_stream.put(2).await;
    spawn(async move {
        while let Some(r) = client_stream.next().await {
            println!("{:?}", r);
        }
    });
    sleep(Duration::from_millis(40000)).await;
    assert_eq!("true", var("STREAM_FINISHED").unwrap());
    server_stream.put(3).await;
    Ok(())
}
